﻿<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">

<head>
	<title xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory">第21章. EPL参考: 数据流</title>
	<link rel="stylesheet" href="css/espertech.css" type="text/css">
	<meta xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" name="generator" content="DocBook XSL-NS Stylesheets V1.74.0">
	<meta xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" http-equiv="Content-Type" content="text/html; charset=UTF-8">
	<link rel="home" href="index.html" title="Esper参考">
	<link rel="up" href="index.html" title="Esper参考">
	<link rel="prev" href="spatial.html" title="第20章. EPL参考: 空间方法和索引">
	<link rel="next" href="extension.html" title="第22章. 集成与扩展">
</head>

<body>
	<p xmlns:d="http://docbook.org/ns/docbook" id="title"><a href="./index.html" class="site_href"><strong>www.espertech.com</strong></a><a href="http://www.espertech.com/esper/esper-documentation/" class="doc_href"><strong>文档</strong></a></p>
	<ul xmlns:d="http://docbook.org/ns/docbook" class="docnav">
		<li class="previous"><a accesskey="p" href="spatial.html"><strong>上一页</strong></a></li>
		<li class="next"><a accesskey="n" href="extension.html"><strong>下一页</strong></a></li>
	</ul>
	<div class="chapter" lang="en-US">
		<div class="titlepage">
			<div>
				<div>
					<h2 class="title"><a id="dataflow"></a>第21章. EPL参考: 数据流</h2>
				</div>
			</div>
		</div>
		<div class="toc">
			<dl>
				<dt><a href="dataflow.html#dataflow-intro">21.1. 简介</a></dt>
				<dt><a href="dataflow.html#dataflow-usage">21.2. 使用方法</a></dt>
				<dd>
					<dl>
						<dt><a href="dataflow.html#dataflow-usage-overview">21.2.1. 概述</a></dt>
						<dt><a href="dataflow.html#dataflow-usage-syntax">21.2.2. 语法</a></dt>
					</dl>
				</dd>
				<dt><a href="dataflow.html#dataflow-reference">21.3. 内置操作符</a></dt>
				<dd>
					<dl>
						<dt><a href="dataflow.html#dataflow-reference-beaconsource">21.3.1. BeaconSource</a></dt>
						<dt><a href="dataflow.html#dataflow-reference-epstatementsource">21.3.2. EPStatementSource</a></dt>
						<dt><a href="dataflow.html#dataflow-reference-eventbussink">21.3.3. EventBusSink</a></dt>
						<dt><a href="dataflow.html#dataflow-reference-eventbussource">21.3.4. EventBusSource</a></dt>
						<dt><a href="dataflow.html#dataflow-reference-filter">21.3.5. Filter</a></dt>
						<dt><a href="dataflow.html#dataflow-reference-logsink">21.3.6. LogSink</a></dt>
						<dt><a href="dataflow.html#dataflow-reference-select">21.3.7. Select</a></dt>
					</dl>
				</dd>
				<dt><a href="dataflow.html#dataflow-api">21.4. API</a></dt>
				<dd>
					<dl>
						<dt><a href="dataflow.html#dataflow-api-declaring">21.4.1. 声明数据流</a></dt>
						<dt><a href="dataflow.html#dataflow-api-instantiating">21.4.2. 实例化数据流</a></dt>
						<dt><a href="dataflow.html#dataflow-api-executing">21.4.3. 执行数据流</a></dt>
						<dt><a href="dataflow.html#dataflow-api-instantiationoptions">21.4.4. 实例化选项</a></dt>
						<dt><a href="dataflow.html#dataflow-api-startcaptive">21.4.5. 启动捕获</a></dt>
						<dt><a href="dataflow.html#dataflow-api-punctuation">21.4.6. 使用标记的数据流标点</a></dt>
						<dt><a href="dataflow.html#dataflow-api-exception">21.4.7. 异常处理</a></dt>
					</dl>
				</dd>
				<dt><a href="dataflow.html#dataflow-examples">21.5. 示例</a></dt>
				<dt><a href="dataflow.html#dataflow-op">21.6. 操作符实现</a></dt>
				<dd>
					<dl>
						<dt><a href="dataflow.html#dataflow-op-source">21.6.1. 作为源的示例操作符</a></dt>
						<dt><a href="dataflow.html#dataflow-op-second">21.6.2. 示例分词器操作符</a></dt>
						<dt><a href="dataflow.html#dataflow-op-third">21.6.3. 示例聚合器操作符</a></dt>
						<dt><a href="dataflow.html#dataflow-op-parameters">21.6.4. 传递操作符参数</a></dt>
					</dl>
				</dd>
			</dl>
		</div>
		<div class="sect1" lang="en-US">
			<div class="titlepage">
				<div>
					<div>
						<h2 class="title"><a id="dataflow-intro"></a>21.1. 简介</h2>
					</div>
				</div>
			</div><a id="d0e51221" class="indexterm"></a>
			<p>
				EPL中的数据流有以下用途:
			</p>
			<div xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="orderedlist">
				<ol>
					<li>
						<p>
							支持数据流编程和基于流的编程。
						</p>
					</li>
					<li>
						<p>
							声明式和运行时可管理的输入输出适配器集成，这些适配器可由EsperIO或应用程序提供。
						</p>
					</li>
					<li>
						<p>
							无需使用事件总线，实现仅对数据流可见的事件和事件类型，以获得性能提升。
						</p>
					</li>
				</ol>
			</div>
			<p>
				数据流操作符通过底层事件对象或包装事件的流进行通信。底层事件对象可以是POJO、Map、对象数组或DOM/XML。包装事件由<code class="literal">EventBean</code>实例表示，这些实例将类型信息与底层事件对象关联起来。
			</p>
			<p>
				有关数据流编程或基于流的编程的更多信息，请参阅<a class="link" href="http://en.wikipedia.org/wiki/Flow-based_programming" target="">维基百科FBP文章</a>。
			</p>
			<p>
				EPL提供了许多有用的内置操作符，可以在图中组合这些操作符来编程数据流。此外，EsperIO提供了预构建的操作符，作为事件的源或接收器。应用程序可以轻松创建和使用自己的数据流操作符。
			</p>
			<p>
				使用数据流，应用程序可以直接向数据流操作符提供事件，而无需使用运行时的事件总线。不使用事件总线(如<code class="literal">EPEventService</code>的<code class="literal">sendEvent<span class="emphasis"><em>Type</em></span></code>方法所示)可以获得性能提升，因为运行时不需要将事件与语句匹配，也不需要将底层事件对象包装在<code class="literal">EventBean</code>实例中。
			</p>
			<p>
				数据流还允许对线程、同步和异步操作进行更细粒度的控制。
			</p>
		</div>
		<div class="sect1" lang="en-US">
			<div class="titlepage">
				<div>
					<div>
						<h2 class="title"><a id="dataflow-usage"></a>21.2. Usage</h2>
					</div>
				</div>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-usage-overview"></a>21.2.1. Overview</h3>
						</div>
					</div>
				</div>
				<p>
					Your application declares a data flow using <code class="literal">create dataflow </code><span class="emphasis"><em>dataflow-name</em></span>. Declaring the data flow causes the EPL compiler to validate the syntax and some aspects of the data flow graph of operators. Declaring the data flow does not actually instantiate or execute a data flow. Resolving event types and instantiating operators (as required) takes place at time of data flow instantiation.
				</p>
				<p>
					After your application has declared a data flow, it can instantiate the data flow and execute it. A data flow can be instantiated as many times as needed and each data flow instance can only be executed once.
				</p>
				<p>
					The example EPL below creates a data flow that, upon execution, outputs the text <code class="literal">Hello World</code> to console and then ends.
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow HelloWorldDataFlow
  BeaconSource -&gt; helloworld.stream { text: 'hello world' , iterations: 1}
  LogSink(helloworld.stream) {}</pre>
				<p>
					The sample data flow above declares a <code class="literal">BeaconSource</code> operator parameterized by the "hello world" text and 1 iteration. The <code class="literal">-&gt;</code> keyword reads as <span class="emphasis"><em>produces streams</em></span>. The <code class="literal">BeaconSource</code> operator produces a single stream named <code class="literal">helloworld.stream</code>. The <code class="literal">LogSink</code> operator receives this stream and prints it unformatted.
				</p>
				<p>
					The next program code snippet declares the data flow to the runtime:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">String epl = "create dataflow HelloWorldDataFlow\n" +
  "BeaconSource -&gt; helloworldStream { text: 'hello world' , iterations: 1}\n" +
  "LogSink(helloworldStream) {}";

Configuration configuration = new Configuration();
CompilerArguments compilerArguments = new CompilerArguments(configuration);
EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, compilerArguments);
EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);</pre>
				<p>
					After declaring a data flow to a runtime, your application can then instantiate and execute the data flow.
				</p>
				<p>
					The following program code snippet instantiates the data flow:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">EPDataFlowInstance instance =
  runtime.getDataFlowService().instantiate(deployment.getDeploymentId(), "HelloWorldDataFlow");</pre>
				<p>
					A data flow instance is represented by an <code class="literal">EPDataFlowInstance</code> object.
				</p>
				<p>
					The next code snippet executes the data flow instance:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">instance.run();</pre>
				<p>
					By using the <code class="literal">run</code> method of <code class="literal">EPDataFlowInstance</code> the runtime executes the data flow using the same thread (blocking execute) and returns when the data flow completes. A data flow completes when all operators receive final markers.
				</p>
				<p>
					The hello world data flow simply prints an unformatted <code class="literal">Hello World</code> string to console. Please check the built-in operator reference for <code class="literal">BeaconSource</code> and <code class="literal">LogSink</code> for more options.
				</p>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-usage-syntax"></a>21.2.2. Syntax</h3>
						</div>
					</div>
				</div>
				<p>
					The synopsis for declaring a data flow is:
				</p>
				<pre class="synopsis">create dataflow <span class="emphasis"><em>name</em></span>
	[<span class="emphasis"><em>schema_declarations</em></span>]
	[<span class="emphasis"><em>operator_declarations</em></span>]
	</pre>
				<p>
					After <code class="literal">create dataflow</code> follows the data flow name and a mixed list of event type (schema) declarations and operator declarations.
				</p>
				<p>
					Schema declarations define an event type. Specify any number of <code class="literal">create schema</code> clauses as part of the data flow declaration followed by a comma character to end each schema declaration. The syntax for <code class="literal">create schema</code> is described in <a class="xref" href="epl_clauses.html#epl_createschema" title="5.15. Declaring an Event Type: Create Schema">Section 5.15, “Declaring an Event Type: Create Schema”</a>.
				</p>
				<p>
					All event types that are defined as part of a data flow are private to the data flow and not available to other statements. To define event types that are available across data flows and other statements, use a <code class="literal">create schema</code> statement, runtime or static configuration.
				</p>
				<p>
					Annotations as well as expression declarations and scripts can also be pre-pended to the data flow declaration.
				</p>
				<div class="sect3" lang="en-US">
					<div class="titlepage">
						<div>
							<div>
								<h4 class="title"><a id="dataflow-usage-syntax-op"></a>21.2.2.1. Operator Declaration</h4>
							</div>
						</div>
					</div>
					<p>
						For each operator, declare the operator name, input streams, output streams and operator parameters.
					</p>
					<p>
						The syntax for declaring a data flow operator is:
					</p>
					<pre class="synopsis"><span class="emphasis"><em>operator_name</em></span> [(<span class="emphasis"><em>input_streams</em></span>)]  [-&gt; <span class="emphasis"><em>output_streams</em></span>] {
  [<span class="emphasis"><em>parameter_name</em></span> : <span class="emphasis"><em>parameter_value_expr</em></span>] [, ...]
}</pre>
					<p>
						The operator name is an identifier that identifies an operator.
					</p>
					<p>
						If the operator accepts input streams then those may be listed in parenthesis after the operator name, see <a class="xref" href="dataflow.html#dataflow-usage-syntax-opinputstreams" title="21.2.2.2. Declaring Input Streams">Section 21.2.2.2, “Declaring Input Streams”</a>.
					</p>
					<p>
						If the operator can produce output streams then specify <code class="literal">-&gt;</code> followed by a list of output stream names and types. See <a class="xref" href="dataflow.html#dataflow-usage-syntax-opoutputstreams" title="21.2.2.3. Declaring Output Streams">Section 21.2.2.3, “Declaring Output Streams”</a>.
					</p>
					<p>
						Following the input and output stream declaration provide curly brackets (<code class="literal">{}</code>) containing operator parameters. See <a class="xref" href="dataflow.html#dataflow-usage-syntax-opparams" title="21.2.2.4. Declaring Operator Parameters">Section 21.2.2.4, “Declaring Operator Parameters”</a>.
					</p>
					<p>
						An operator that receives no input streams, produces no output streams and has no parameters assigned to it is shown in this EPL example data flow:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperatorSimple {}</pre>
					<p>
						The next EPL shows a data flow that consists of an operator <code class="literal">MyOperator</code> that receives a single input stream <code class="literal">myInStream</code> and produces a single output stream <code class="literal">myOutStream</code> holding <code class="literal">MyEvent</code> events.
						The EPL configures the operator parameter <code class="literal">myParameter</code> with a value of 10:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create schema MyEvent as (id string, price double),
  MyOperator(myInStream) -&gt; myOutStream&lt;MyEvent&gt; {
    myParameter : 10
  }</pre>
					<p>
						The next sections outline input stream, output stream and parameter assignment in greater detail.
					</p>
				</div>
				<div class="sect3" lang="en-US">
					<div class="titlepage">
						<div>
							<div>
								<h4 class="title"><a id="dataflow-usage-syntax-opinputstreams"></a>21.2.2.2. Declaring Input Streams</h4>
							</div>
						</div>
					</div>
					<p>
						In case the operator receives input streams, list the input stream names within parenthesis following the operator name. As part of the input stream declaration you may use the <code class="literal">as</code> keyword to assign an alias short name to one or multiple input streams.
					</p>
					<p>
						The EPL shown next declares <code class="literal">myInStream</code> and assigns the alias <code class="literal">mis</code>:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator(myInStream as mis) {}</pre>
					<p>
						Multiple input streams can be listed separated by comma. We use the term <span class="emphasis"><em>input port</em></span> to mean the ordinal number of the input stream in the order the input streams are listed.
					</p>
					<p>
						The EPL as below declares two input streams and assigns an alias to each. The runtime assigns <code class="literal">streamOne</code> to input port 0 (zero) and <code class="literal">streamTwo</code> to port 1.
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator(streamOne as one, streamTwo as two) {}</pre>
					<p>
						You may assign multiple input streams to the same port and alias by placing the stream names into parenthesis. All input streams for the same port must have the same event type associated.
					</p>
					<p>
						The next statement declares an operator that receives input streams <code class="literal">streamA</code> and <code class="literal">streamB</code> both assigned to port 0 (zero) and alias <code class="literal">streamsAB</code>:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator( (streamA, streamB) as streamsAB) {}</pre>
					<p>
						Input and output stream names can have the dot-character in their name.
					</p>
					<p>
						The following is also valid EPL:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator(my.in.stream) -&gt; my.out.stream {}</pre>
					<div xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="note">
						<h2>Note</h2>
						<p>
							Reserved keywords may not appear in the stream name.
						</p>
					</div>
				</div>
				<div class="sect3" lang="en-US">
					<div class="titlepage">
						<div>
							<div>
								<h4 class="title"><a id="dataflow-usage-syntax-opoutputstreams"></a>21.2.2.3. Declaring Output Streams</h4>
							</div>
						</div>
					</div>
					<p>
						In case the operator produces output streams, list the output streams after the <code class="literal">-&gt;</code> keyword. Multiple output streams can be listed separated by comma. We use the term <span class="emphasis"><em>output port</em></span> to mean the ordinal number of the output stream in the order the output streams are listed.
					</p>
					<p>
						The sample EPL below declares an operator that produces two output streams <code class="literal">my.out.one</code> and <code class="literal">my.out.two</code>.
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator -&gt; my.out.one, my.out.two {}</pre>
					<p>
						Each output stream can be assigned optional type information within less/greater-then (<code class="literal">&lt;&gt;</code>). Type information is required if the operator cannot deduce the output type from the input type and the operator does not declare explicit output type(s). The event type name can
						either be an event type defined within the same data flow or an event type defined in the runtime.
					</p>
					<p>
						This EPL example declares an <code class="literal">RFIDSchema</code> event type based on an object-array event representation and associates the output stream <code class="literal">rfid.stream</code> with the <code class="literal">RFIDSchema</code> type.
						The stream <code class="literal">rfid.stream</code> therefore carries object-array (<code class="literal">Object[]</code>) typed objects according to schema <code class="literal">RFIDSchema</code>:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create objectarray schema RFIDSchema (tagId string, locX double, locY double),
  MyOperator -&gt; rfid.stream&lt;RFIDSchema&gt; {}</pre>
					<p>
						The keyword <code class="literal">eventbean</code> is reserved: Use <code class="literal">eventbean&lt;</code><span class="emphasis"><em>type-name</em></span><code class="literal">&gt;</code> to indicate that a stream carries <code class="literal">EventBean</code> instances of the given type instead of the underlying event object.
					</p>
					<p>
						This EPL example declares an <code class="literal">RFIDSchema</code> event type based on an object-array event representation and associates the output stream <code class="literal">rfid.stream</code> with the event type,
						such that the stream <code class="literal">rfid.stream</code> carries <code class="literal">EventBean</code> objects:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create objectarray schema RFIDSchema (tagId string, locX double, locy double),
  MyOperator -&gt; rfid.stream&lt;eventbean&lt;RFIDSchema&gt;&gt; {}</pre>
					<p>
						Use questionmark (<code class="literal">?</code>) to indicate that the type of events is not known in advance.
					</p>
					<p>
						In the next EPL the stream <code class="literal">my.stream</code> carries <code class="literal">EventBean</code> instances of any type:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator -&gt; my.stream&lt;eventbean&lt;?&gt;&gt; {}</pre>
				</div>
				<div class="sect3" lang="en-US">
					<div class="titlepage">
						<div>
							<div>
								<h4 class="title"><a id="dataflow-usage-syntax-opparams"></a>21.2.2.4. Declaring Operator Parameters</h4>
							</div>
						</div>
					</div>
					<p>
						Operators can receive constants, objects, EPL expressions and complete statements as parameters. All parameters are listed within curly brackets (<code class="literal">{}</code>) after input and output stream declarations. Curly brackets are required as a separator even if
						the operator has no parameters.
					</p>
					<p>
						The syntax for parameters is:
					</p>
					<pre class="synopsis"><span class="emphasis"><em>name</em></span> : <span class="emphasis"><em>value_expr</em></span> [,...]</pre>
					<p>
						The parameter name is an identifier that is followed by the colon (<code class="literal">:</code>) or equals (<code class="literal">=</code>) character and a value expression. A value expression can be any expression, system property, JSON notation object or statement. Parameters are separated by comma character.
					</p>
					<p>
						The next EPL demonstrates operator parameters that are scalar values:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator {
    stringParam : 'sample',
    secondString : "double-quotes are fine",
    intParam : 10
  }</pre>
					<p>
						Operator parameters can be any EPL expression including expressions that use variables. Subqueries, aggregations and the <code class="literal">prev</code> and <code class="literal">prior</code> functions cannot be applied here.
					</p>
					<p>
						The EPL shown below lists operator parameters that are expressions:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator {
    intParam : 24*60*60,
    threshold : var_threshold	// a variable defined in the runtime
  }</pre>
					<p>
						To obtain the value of a system property, the special <code class="literal">systemProperties</code> property name is reserved for access to system properties.
					</p>
					<p>
						The following EPL sets operator parameters to a value obtained from a system property:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator {
    someSystemProperty : systemProperties('mySystemProperty') 
  }</pre>
					<p>
						Any JSON value can also be used as a value. Use square brackets <code class="literal">[]</code> for JSON arrays. Use curly brackets <code class="literal">{}</code> to hold nested Map or other object values. Provide the special <code class="literal">class</code> property to
						instantiate a given instance by class name. The runtime populates the respective array, Map or Object as specified in the JSON parameter value.
					</p>
					<p>
						The below EPL demonstrates operator parameters that are JSON values:
					</p>
					<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  MyOperator {
    myStringArray: ['a', "b"],
    myMapOrObject: {
      a : 10,
      b : 'xyz',
    },
    myInstance: {
      class: 'com.myorg.myapp.MyImplementation',
      myValue : 'sample'
    }
  }</pre>
					<p>
						The special parameter name <code class="literal">select</code> is reserved for use with EPL select statements. Please see the <code class="literal">Select</code> built-in operator for an example.
					</p>
				</div>
			</div>
		</div>
		<div class="sect1" lang="en-US">
			<div class="titlepage">
				<div>
					<div>
						<h2 class="title"><a id="dataflow-reference"></a>21.3. Built-In Operators</h2>
					</div>
				</div>
			</div>
			<p>
				The below table summarizes the built-in data flow operators available:
			</p>
			<div class="table"><a id="data_flow_operators_esper"></a>
				<p class="title"><b>Table 21.1. Built-in Operators</b></p>
				<div class="table-contents">
					<table summary="Built-in Operators" border="1">
						<colgroup>
							<col>
							<col>
						</colgroup>
						<thead>
							<tr>
								<th>Operator</th>
								<th>Description</th>
							</tr>
						</thead>
						<tbody>
							<tr>
								<td>BeaconSource</td>
								<td>
									<p>
										Utility source that generates events. See <a class="xref" href="dataflow.html#dataflow-reference-beaconsource" title="21.3.1. BeaconSource">Section 21.3.1, “BeaconSource”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>Emitter</td>
								<td>
									<p>
										Special operator for injecting events into a stream. See <a class="xref" href="dataflow.html#dataflow-api-startcaptive" title="21.4.5. Start Captive">Section 21.4.5, “Start Captive”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>EPStatementSource</td>
								<td>
									<p>
										One or more statements act as event sources. See <a class="xref" href="dataflow.html#dataflow-reference-epstatementsource" title="21.3.2. EPStatementSource">Section 21.3.2, “EPStatementSource”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>EventBusSink</td>
								<td>
									<p>
										The event bus is the sink: Sends events from the data flow into the event bus. See <a class="xref" href="dataflow.html#dataflow-reference-eventbussink" title="21.3.3. EventBusSink">Section 21.3.3, “EventBusSink”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>EventBusSource</td>
								<td>
									<p>
										The event bus is the source: Receives events from the event bus into the data flow. See <a class="xref" href="dataflow.html#dataflow-reference-eventbussource" title="21.3.4. EventBusSource">Section 21.3.4, “EventBusSource”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>Filter</td>
								<td>
									<p>
										Filters an input stream and produces an output stream containing the events passing the filter criteria. See <a class="xref" href="dataflow.html#dataflow-reference-filter" title="21.3.5. Filter">Section 21.3.5, “Filter”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>LogSink</td>
								<td>
									<p>
										Utility sink that outputs events to console or log. See <a class="xref" href="dataflow.html#dataflow-reference-logsink" title="21.3.6. LogSink">Section 21.3.6, “LogSink”</a>.
									</p>
								</td>
							</tr>
							<tr>
								<td>Select</td>
								<td>
									<p>
										An EPL select statement that executes on the input stream events. See <a class="xref" href="dataflow.html#dataflow-reference-select" title="21.3.7. Select">Section 21.3.7, “Select”</a>.
									</p>
								</td>
							</tr>
						</tbody>
					</table>
				</div>
			</div><br class="table-break">
			<p>
				The below table summarizes the built-in EsperIO data flow operators. Please see the EsperIO documentation and source for more information.
			</p>
			<div class="table"><a id="data_flow_operators_esperio"></a>
				<p class="title"><b>Table 21.2. EsperIO Built-in Operators</b></p>
				<div class="table-contents">
					<table summary="EsperIO Built-in Operators" border="1">
						<colgroup>
							<col>
							<col>
						</colgroup>
						<thead>
							<tr>
								<th>Operator</th>
								<th>Description</th>
							</tr>
						</thead>
						<tbody>
							<tr>
								<td>AMQPSource</td>
								<td>
									<p>
										Attaches to AMQP broker to receive messages to process.
									</p>
								</td>
							</tr>
							<tr>
								<td>AMQPSink</td>
								<td>
									<p>
										Attaches to AMQP broker to send messages.
									</p>
								</td>
							</tr>
							<tr>
								<td>FileSource</td>
								<td>
									<p>
										Reads one or more files and produces events from file data.
									</p>
								</td>
							</tr>
							<tr>
								<td>FileSink</td>
								<td>
									<p>
										Write one or more files from events received.
									</p>
								</td>
							</tr>
						</tbody>
					</table>
				</div>
			</div><br class="table-break">
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-beaconsource"></a>21.3.1. BeaconSource</h3>
						</div>
					</div>
				</div>
				<p>
					The BeaconSource operator generates events and populates event properties.
				</p>
				<p>
					The BeaconSource operator does not accept any input streams and has no input ports.
				</p>
				<p>
					The BeaconSource operator must have a single output stream. When the BeaconSource operator completed generating events according to the number of iterations provided or when it is cancelled it outputs a final marker to the output stream.
				</p>
				<p>
					Parameters for the BeaconSource operator are all optional parameters:
				</p>
				<div class="table"><a id="beacon_params"></a>
					<p class="title"><b>Table 21.3. BeaconSource Parameters</b></p>
					<div class="table-contents">
						<table summary="BeaconSource Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>initialDelay</td>
									<td>Specifies the number of seconds delay before producing events.</td>
								</tr>
								<tr>
									<td>interval</td>
									<td>Time interval between events. Takes a integer or double-typed value for the number of seconds. The interval is zero when not provided.</td>
								</tr>
								<tr>
									<td>iterations</td>
									<td>Number of events produced. Takes an integer value. When not provided the operator produces tuples until the data flow instance gets cancelled.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					Event properties to be populated can simply be added to the parameters.
				</p>
				<p>
					If your declaration provides an event type for the output stream then BeaconSource will populate event properties of the underlying events. If no event type is specified, BeaconSource creates an anonymous object-array event type to carry the event properties that are generated and associates this type with its output stream.
				</p>
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create schema SampleSchema(tagId string, locX double),	// sample type			
			
  // BeaconSource that produces empty object-array events without delay 
  // or interval until cancelled.
  BeaconSource -&gt; stream.one {}
  
  // BeaconSource that produces one RFIDSchema event populating event properties
  // from a user-defined function "generateTagId" and the provided values.
  BeaconSource -&gt; stream.two&lt;SampleSchema&gt; {
    iterations : 1,
    tagId : generateTagId(),
    locX : 10
  }
  
  // BeaconSource that produces 10 object-array events populating
  // the price property with a random value.
  BeaconSource -&gt; stream.three {
    iterations : 10,
    interval : 10, // every 10 seconds
    initialDelay : 5, // start after 5 seconds
    price : Math.random() * 100
  }</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-epstatementsource"></a>21.3.2. EPStatementSource</h3>
						</div>
					</div>
				</div>
				<p>
					The EPStatementSource operator maintains a subscription to the results of one or more statements. The operator produces the statement output events.
				</p>
				<p>
					The EPStatementSource operator does not accept any input streams and has no input ports.
				</p>
				<p>
					The EPStatementSource operator must have a single output stream. It does not generate a final or other marker.
				</p>
				<p>
					Either the statement name or the statement filter parameter is required:
				</p>
				<div class="table"><a id="epstatementsource_params"></a>
					<p class="title"><b>Table 21.4. EPStatementSource Parameters</b></p>
					<div class="table-contents">
						<table summary="EPStatementSource Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>collector</td>
									<td>Optional parameter, used to transform statement output events to submitted events.</td>
								</tr>
								<tr>
									<td>statementName</td>
									<td>Name of the statement that produces events. The statement does not need to exist at the time of data flow instantiation.</td>
								</tr>
								<tr>
									<td>statementFilter</td>
									<td>Implementation of the <code class="literal">EPDataFlowEPStatementFilter</code> that returns true for each statement that produces events. Statements do not need to exist at the time of data flow instantiation.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					If a statement name is provided, the operator subscribes to output events of the statement if the statement exists or when it gets created at a later point in time.
				</p>
				<p>
					If a statement filter is provided instead, the operator subscribes to output events of all statements that currently exist and pass the filter <code class="literal">pass</code> method or that get created at a later point in time and pass the filter <code class="literal">pass</code> method.
				</p>
				<p>
					The <code class="literal">collector</code> can be specified to transform output events. If no collector is specified the operator submits the underlying events of the insert stream received from the statement. The collector object must implement the interface <code class="literal">EPDataFlowIRStreamCollector</code>.
				</p>
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create schema SampleSchema(tagId string, locX double),	// sample type			
			
  // Consider only the statement named MySelectStatement when it exists.
  // No transformation.
  EPStatementSource -&gt; stream.one&lt;eventbean&lt;?&gt;&gt; {
    statementName : 'MySelectStatement'
  }
  
  // Consider all statements that match the filter object provided.
  // No transformation.
  EPStatementSource -&gt; stream.two&lt;eventbean&lt;?&gt;&gt; {
    statementFilter : {
      class : 'com.mycompany.filters.MyStatementFilter'
    }
  }
  
  // Consider all statements that match the filter object provided.
  // With collector that performs transformation.
  EPStatementSource -&gt; stream.two&lt;SampleSchema&gt; {
    collector : {
      class : 'com.mycompany.filters.MyCollector'
    },
    statementFilter : {
      class : 'com.mycompany.filters.MyStatementFilter'
    }
  }</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-eventbussink"></a>21.3.3. EventBusSink</h3>
						</div>
					</div>
				</div>
				<p>
					The EventBusSink operator send events received from a data flow into the event bus. Any statement that looks for any of the events gets triggered, equivalent to the <code class="literal">sendEvent<span class="emphasis"><em>Type</em></span></code> methods on <code class="literal">EPEventService</code> or the <code class="literal">insert into</code> clause.
				</p>
				<p>
					The EventBusSink operator accepts any number of input streams. The operator forwards all events arriving on any input ports to the event bus, equivalent to the <code class="literal">sendEvent<span class="emphasis"><em>Type</em></span></code> methods on <code class="literal">EPEventService</code>.
				</p>
				<p>
					The EventBusSink operator cannot declare any output streams.
				</p>
				<p>
					Parameters for the EventBusSink operator are all optional parameters:
				</p>
				<div class="table"><a id="eventbussink_params"></a>
					<p class="title"><b>Table 21.5. EventBusSink Parameters</b></p>
					<div class="table-contents">
						<table summary="EventBusSink Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>collector</td>
									<td>Optional parameter, used to transform data flow events to event bus events.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					The <code class="literal">collector</code> can be specified to transform data flow events to event bus events. If no collector is specified the operator submits the events directly to the event bus. The collector object must implement the interface <code class="literal">EPDataFlowEventCollector</code>.
				</p>
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  BeaconSource -&gt; instream&lt;SampleSchema&gt; {}  // produces a sample stream
  
  // Send SampleSchema events produced by beacon to the event bus.
  EventBusSink(instream) {}
  
  // Send SampleSchema events produced by beacon to the event bus.
  // With collector that performs transformation.
  EventBusSink(instream) {
    collector : {
      class : 'com.mycompany.filters.MyCollector'
    }
  }</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-eventbussource"></a>21.3.4. EventBusSource</h3>
						</div>
					</div>
				</div>
				<p>
					The EventBusSource operator receives events from the event bus and produces an output stream of the events received. With the term event bus we mean any event visible to the runtime either because the application send the event via any of the <code class="literal">sendEvent<span class="emphasis"><em>Type</em></span></code> methods on <code class="literal">EPEventService</code> or because statements populated streams as a result of <code class="literal">insert into</code>.
				</p>
				<p>
					The EventBusSource operator does not accept any input streams and has no input ports.
				</p>
				<p>
					The EventBusSource operator must have a single output stream. It does not generate a final or other marker. The event type declared for the output stream is the event type of events received from the event bus.
				</p>
				<p>
					All parameters to EventBusSource are optional:
				</p>
				<div class="table"><a id="eventbussource_params"></a>
					<p class="title"><b>Table 21.6. EventBusSource Parameters</b></p>
					<div class="table-contents">
						<table summary="EventBusSource Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>collector</td>
									<td>Optional parameter and used to transform event bus events to submitted events.</td>
								</tr>
								<tr>
									<td>filter</td>
									<td>Filter expression for event bus matching.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					The <code class="literal">collector</code> can be specified to transform output events. If no collector is specified the operator submits the underlying events of the stream received from the event bus. The collector object must implement the interface <code class="literal">EPDataFlowEventBeanCollector</code>.
				</p>
				<p>
					The <code class="literal">filter</code> is an expression that the event bus compiles and efficiently matches even in the presence of a large number of event bus sources. The filter expression must return a boolean-typed value, returning true for those events that the event bus passes to the operator.
				</p>
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow

  // Receive all SampleSchema events from the event bus.
  // No transformation.
  EventBusSource -&gt; stream.one&lt;SampleSchema&gt; {}
  
  // Receive all SampleSchema events with tag id '001' from the event bus.
  // No transformation.
  EventBusSource -&gt; stream.one&lt;SampleSchema&gt; {
    filter : tagId = '001'
  }

  // Receive all SampleSchema events from the event bus.
  // With collector that performs transformation.
  EventBusSource -&gt; stream.two&lt;SampleSchema&gt; {
    collector : {
      class : 'com.mycompany.filters.MyCollector'
    },
  }</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-filter"></a>21.3.5. Filter</h3>
						</div>
					</div>
				</div>
				<p>
					The Filter operator filters an input stream and produces an output stream containing the events passing the filter criteria. If a second output stream is provided, the operator sends events not passing filter criteria to that output stream.
				</p>
				<p>
					The Filter operator accepts a single input stream.
				</p>
				<p>
					The Filter operator requires one or two output streams. The event type of the input and output stream(s) must be the same. The first output stream receives the matching events according to the filter expression. If declaring two output streams, the second stream receives non-matching events.
				</p>
				<p>
					The Filter operator has a single required parameter:
				</p>
				<div class="table"><a id="filter_params"></a>
					<p class="title"><b>Table 21.7. Filter Parameters</b></p>
					<div class="table-contents">
						<table summary="Filter Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>filter</td>
									<td>The filter criteria expression.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create schema SampleSchema(tagId string, locX double),	// sample type
  BeaconSource -&gt; samplestream&lt;SampleSchema&gt; {}  // sample source
  
  // Filter all events that have a tag id of '001'
  Filter(samplestream) -&gt; tags_001 {
    filter : tagId = '001' 
  }
  
  // Filter all events that have a tag id of '001', 
  // putting all other events into the second stream
  Filter(samplestream) -&gt; tags_001, tags_other {
    filter : tagId = '001' 
  }</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-logsink"></a>21.3.6. LogSink</h3>
						</div>
					</div>
				</div>
				<p>
					The LogSink operator outputs events to console or log file in either a JSON, XML or built-in format (the default).
				</p>
				<p>
					The LogSink operator accepts any number of input streams. All events arriving on any input ports are logged.
				</p>
				<p>
					The LogSink operator cannot declare any output streams.
				</p>
				<p>
					Parameters for the LogSink operator are all optional parameters:
				</p>
				<div class="table"><a id="logsink_params"></a>
					<p class="title"><b>Table 21.8. LogSink Parameters</b></p>
					<div class="table-contents">
						<table summary="LogSink Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>format</td>
									<td>Specify format as a string value: <code class="literal">json</code> for JSON-formatted output, <code class="literal">xml</code> for XML-formatted output and <code class="literal">summary</code> (default) for a built-in format.</td>
								</tr>
								<tr>
									<td>layout</td>
									<td>Pattern string according to which output is formatted. Place <code class="literal">%df</code> for data flow name, <code class="literal">%p</code> for port number, <code class="literal">%i</code> for data flow instance id, <code class="literal">%t</code> for title, <code class="literal">%e</code> for event data.</td>
								</tr>
								<tr>
									<td>log</td>
									<td>Boolean true (default) for log output, false for console output.</td>
								</tr>
								<tr>
									<td>linefeed</td>
									<td>Boolean true (default) for line feed, false for no line feed.</td>
								</tr>
								<tr>
									<td>title</td>
									<td>String title text pre-pended to output.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  BeaconSource -&gt; instream {}  // produces sample stream to use below
  
  // Output textual event to log using defaults.
  LogSink(instream) {}
  
  // Output JSON-formatted to console.
  LogSink(instream) {
    format : 'json',
    layout : '%t [%e]',
    log : false,
    linefeed : true,
    title : 'My Custom Title:'
  }</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-reference-select"></a>21.3.7. Select</h3>
						</div>
					</div>
				</div>
				<p>
					The Select operator is configured with an EPL select statement. It applies events from input streams to the select statement and outputs results either continuously or when the final marker arrives.
				</p>
				<p>
					The Select operator accepts one or more input streams.
				</p>
				<p>
					The Select operator requires a single output stream.
				</p>
				<p>
					The Select operator requires the <code class="literal">select</code> parameter, all other parameters are optional:
				</p>
				<div class="table"><a id="select_params"></a>
					<p class="title"><b>Table 21.9. Select Operator Parameters</b></p>
					<div class="table-contents">
						<table summary="Select Operator Parameters" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>Name</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>iterate</td>
									<td>Boolean indicator whether results should be output continuously or only upon arrival of the final marker.</td>
								</tr>
								<tr>
									<td>select</td>
									<td>EPL <code class="literal">select</code> statement in parenthesis.</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
				<p>
					Set the optional <code class="literal">iterate</code> flag to false (the default) to have the operator output results continuously. Set the <code class="literal">iterate</code> flag to true to indicate that the operator outputs results only when the final marker arrives. If <code class="literal">iterate</code> is true then output rate limiting clauses are not supported.
				</p>
				<p>
					The <code class="literal">select</code> parameter is required and provides an EPL select statement within parenthesis.
					For each input port the statement should list the input stream name or the alias name in the <code class="literal">from</code> clause. Only filter-based streams are allowed in the <code class="literal">from</code> clause and patterns or named windows are not supported. Also not allowed are
					the <code class="literal">insert into</code> clause, the <code class="literal">irstream</code> keyword and subselects.
				</p>
				<p>
					The Select operator determines the event type of output events based on the <code class="literal">select</code> clause. It is not necessary to declare an event type for the output stream.
				</p>
				<p>
					Examples are:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow MyDataFlow
  create schema SampleSchema(tagId string, locX double),	// sample type			
  BeaconSource -&gt; instream&lt;SampleSchema&gt; {}  // sample stream
  BeaconSource -&gt; secondstream&lt;SampleSchema&gt; {}  // sample stream
  
  // Simple continuous count of events
  Select(instream) -&gt; outstream {
    select: (select count(*) from instream)
  }
  
  // Demonstrate use of alias
  Select(instream as myalias) -&gt; outstream {
    select: (select count(*) from myalias)
  }
  
  // Output only when the final marker arrives
  Select(instream as myalias) -&gt; outstream {
    select: (select count(*) from myalias),
    iterate: true
  }

  // Same input port for the two sample streams
  Select( (instream, secondstream) as myalias) -&gt; outstream {
    select: (select count(*) from myalias)
  }

  // A join with multiple input streams,
  // joining the last event per stream forming pairs
  Select(instream, secondstream) -&gt; outstream {
    select: (select a.tagId, b.tagId 
        from instream#lastevent as a, secondstream#lastevent as b)
  }
  
  // A join with multiple input streams and using aliases.
  Select(instream as S1, secondstream as S2) -&gt; outstream {
    select: (select a.tagId, b.tagId 
        from S1#lastevent as a, S2#lastevent as b)
  }</pre>
			</div>
		</div>
		<div class="sect1" lang="en-US">
			<div class="titlepage">
				<div>
					<div>
						<h2 class="title"><a id="dataflow-api"></a>21.4. API</h2>
					</div>
				</div>
			</div>
			<p>
				This section outlines the steps to declare, instantiate, execute and cancel or complete data flows.
			</p>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-declaring"></a>21.4.1. Declaring a Data Flow</h3>
						</div>
					</div>
				</div>
				<p>
					Compile data flow the same as any other statement and deploy the compiled module. The <code class="literal">EPStatementObjectModel</code> statement object model can also be used to compile a data flow.
				</p>
				<p>
					Annotations that are listed at the top of the EPL text are applied to all statements and operators in the data flow. Annotations listed for a specific operator apply to that operator only.
				</p>
				<p>
					The next program code snippet declares a data flow to the runtime:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">String epl = "@Name('MyStatementName') create dataflow HelloWorldDataFlow\n" +
  "BeaconSource -&gt; helloworldStream { text: 'hello world' , iterations: 1}\n" +
  "LogSink(helloworldStream) {}";

Configuration configuration = new Configuration();
CompilerArguments compilerArguments = new CompilerArguments(configuration);
EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, compilerArguments);
EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);</pre>
				<p>
					The statement name that can be assigned to the statement is used only for statement management. Your application may undeploy the statement declaring the data flow thereby making the data flow unavailable for instantiation. Existing instances of the data flow are not affected by an undeploy of the statement that declares the data flow.
				</p>
				<p>
					Listeners or the subscriber to the statement declaring a data flow receive no events or other output. The statement declaring a data flow returns no rows when iterated.
				</p>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-instantiating"></a>21.4.2. Instantiating a Data Flow</h3>
						</div>
					</div>
				</div>
				<p>
					The <code class="literal">com.espertech.esper.common.client.dataflow.core.EPDataFlowService</code> available via <code class="literal">getDataFlowService</code> on <code class="literal">EPRuntime</code> manages declared data flows.
				</p>
				<p>
					Use the <code class="literal">instantiate</code> method on <code class="literal">EPDataFlowRuntime</code> to instantiate a data flow after it has been declared. Pass the data flow name and optional instantiation options to the method. A data flow can be instantiated any number of times.
				</p>
				<p>
					A data flow instance is represented by an instance of <code class="literal">EPDataFlowInstance</code>. Each instance has a state as well as methods to start, run, join and cancel as well as methods to obtain execution statistics.
				</p>
				<p>
					Various optional arguments including operator parameters can be passed to <code class="literal">instantiate</code> via the <code class="literal">EPDataFlowInstantiationOptions</code> object as explained in more detail below.
				</p>
				<p>
					The following code snippet instantiates the data flow:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">EPDataFlowInstance instance =
  runtime.getDataFlowService().instantiate(deployment.getDeploymentId(), "HelloWorldDataFlow");</pre>
				<p>
					The runtime does not track or otherwise retain data flow instances in memory. It is up to your application to retain data flow instances as needed.
				</p>
				<p>
					Each data flow instance associates to a state. The start state is <code class="literal">EPDataFlowState.INSTANTIATED</code>. The end state is either <code class="literal">COMPLETED</code> or <code class="literal">CANCELLED</code>.
				</p>
				<p>
					The following table outlines all states:
				</p>
				<div class="table"><a id="data_flow_instance_states"></a>
					<p class="title"><b>Table 21.10. Data Flow Instance States</b></p>
					<div class="table-contents">
						<table summary="Data Flow Instance States" border="1">
							<colgroup>
								<col>
								<col>
							</colgroup>
							<thead>
								<tr>
									<th>State</th>
									<th>Description</th>
								</tr>
							</thead>
							<tbody>
								<tr>
									<td>INSTANTIATED</td>
									<td>
										<p>
											Start state, applies when a data flow instance has been instantiated and has not executed.
										</p>
									</td>
								</tr>
								<tr>
									<td>RUNNING</td>
									<td>
										<p>
											A data flow instance transitions from instantiated to running when any of the <code class="literal">start</code>, <code class="literal">run</code> or <code class="literal">startCaptive</code> methods are invoked.
										</p>
									</td>
								</tr>
								<tr>
									<td>COMPLETED</td>
									<td>
										<p>
											A data flow instance transitions from running to completed when all final markers have been processed by all operators.
										</p>
									</td>
								</tr>
								<tr>
									<td>CANCELLED</td>
									<td>
										<p>
											A data flow instance transitions from running to cancelled when your application invokes the <code class="literal">cancel</code> method on the data flow instance.
										</p>
									</td>
								</tr>
							</tbody>
						</table>
					</div>
				</div><br class="table-break">
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-executing"></a>21.4.3. Executing a Data Flow</h3>
						</div>
					</div>
				</div>
				<p>
					After your application instantiated a data flow instance it can execute the data flow instance using either the <code class="literal">start</code>, <code class="literal">run</code> or <code class="literal">startCaptive</code> methods.
				</p>
				<p>
					Use the <code class="literal">start</code> method to have the runtime allocate a thread for each source operator. Execution is non-blocking. Use the <code class="literal">join</code> method to have one or more threads join a data flow instance execution.
				</p>
				<p>
					Use the <code class="literal">run</code> method to have the runtime use the current thread to execute the single source operator. Multiple source operators are not allowed when using <code class="literal">run</code>.
				</p>
				<p>
					Use the <code class="literal">startCaptive</code> method to have the runtime return all <code class="literal">Runnable</code> instances and emitters, for the purpose of having complete control over execution. The runtime allocates no threads and does not perform any logic for the data flow unless your application employs the <code class="literal">Runnable</code> instances and emitters returned by the method.
				</p>
				<p>
					The next code snippet executes the data flow instance as a blocking call:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">instance.run();</pre>
				<p>
					By using the <code class="literal">run</code> method of <code class="literal">EPDataFlowInstance</code> the runtime executes the data flow instance using the same thread (blocking execute) and returns when the data flow instance completes. A data flow instance completes when all operators receive final markers.
				</p>
				<p>
					The hello world data flow simply prints an unformatted <code class="literal">Hello World</code> string to console. The <code class="literal">BeaconSource</code> operator generates a final marker when it finishes the 1 iteration. The data flow instance thus transitions to complete after the <code class="literal">LogSink</code> operator receives the final marker, and the thread invoking the <code class="literal">run</code> method returns.
				</p>
				<p>
					The next code snippet executes the data flow instance as a non-blocking call:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">instance.start();</pre>
				<p>
					Use the <code class="literal">cancel</code> method to cancel execution of a running data flow instance:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">instance.cancel();</pre>
				<p>
					Use the <code class="literal">join</code> method to join execution of a running data flow instance, causing the joining thread to block until the data flow instance either completes or is cancelled:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">instance.join();</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-instantiationoptions"></a>21.4.4. Instantiation Options</h3>
						</div>
					</div>
				</div>
				<p>
					The <code class="literal">EPDataFlowInstantiationOptions</code> object that can be passed to the <code class="literal">instantiate</code> method may be used to customize the operator graph, operator parameters and execution of the data flow instance.
				</p>
				<p>
					Passing runtime parameters to data flow operators is easiest using the <code class="literal">addParameterURI</code> method. The first parameter is the data flow operator name and the
					operator parameter name separated by the slash character. The second parameter is the value object.
				</p>
				<p>
					For example, in order to pass the file name to the <code class="literal">FileSource</code> operator at runtime, use the following code:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">EPDataFlowInstantiationOptions options = new EPDataFlowInstantiationOptions();
options.addParameterURI("FileSource/file", filename);
EPDataFlowInstance instance = runtime.getDataFlowService().instantiate(deployment.getDeploymentId(), "MyFileReaderDataFlow",options);
instance.run();</pre>
				<p>
					The optional <code class="literal">operatorProvider</code> member takes an implementation of the <code class="literal">EPDataFlowOperatorProvider</code> interface. The runtime invokes this provider to obtain operator instances.
				</p>
				<p>
					The optional <code class="literal">parameterProvider</code> member takes an implementation of the <code class="literal">EPDataFlowOperatorParameterProvider</code> interface. The runtime invokes this provider to obtain operator parameter values. The values override the values provided via parameter URI above.
				</p>
				<p>
					The optional <code class="literal">exceptionHandler</code> member takes an implementation of the <code class="literal">EPDataFlowExceptionHandler</code> interface. The runtime invokes this provider to when exceptions occur.
				</p>
				<p>
					The optional <code class="literal">dataFlowInstanceId</code> can be assigned any string value for the purpose of identifying the data flow instance.
				</p>
				<p>
					The optional <code class="literal">dataFlowInstanceUserObject</code> can be assigned any object value for the purpose of associating a user object to the data flow instance.
				</p>
				<p>
					Set the <code class="literal">operatorStatistics</code> flag to true to obtain statistics for operator execution.
				</p>
				<p>
					Set the <code class="literal">cpuStatistics</code> flag to true to obtain CPU statistics for operator execution.
				</p>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-startcaptive"></a>21.4.5. Start Captive</h3>
						</div>
					</div>
				</div>
				<p>
					Use the <code class="literal">startCaptive</code> method on a <code class="literal">EPDataFlowInstance</code> data flow instance when your application requires full control over threading. This method returns an <code class="literal">EPDataFlowInstanceCaptive</code> instance
					that contains a list of <code class="literal">java.lang.Runnable</code> instances that represent each source operator.
				</p>
				<p>
					The special <code class="literal">Emitter</code> operator can occur in a data flow. This emitter can be used to inject events into the data flow without writing a new operator. Emitter takes a single <code class="literal">name</code> parameter that provides the name of the emitter
					and that is returned in a map of emitters by <code class="literal">EPDataFlowInstanceCaptive</code>.
				</p>
				<p>
					The example EPL below creates a data flow that uses emitter.
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow HelloWorldDataFlow
  create objectarray schema SampleSchema(text string),	// sample type		
	
  Emitter -&gt; helloworld.stream&lt;SampleSchema&gt; { name: 'myemitter' }
  LogSink(helloworld.stream) {}</pre>
				<p>
					Your application may obtain the Emitter instance and sends events directly into the output stream. This feature is only supported in relationship with <code class="literal">startCaptive</code> since the runtime does not allocate any threads or run source operators.
				</p>
				<p>
					The example code snippet below obtains the emitter instance and send events directly into the data flow instance:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">EPDataFlowInstance instance =
      runtime.getDataFlowService().instantiate(deployment.getDeploymentId(), "HelloWorldDataFlow", options);
EPDataFlowInstanceCaptive captiveStart = instance.startCaptive();
Emitter emitter = captiveStart.getEmitters().get("myemitter");
emitter.submit(new Object[] {"this is some text"});</pre>
				<p>
					When emitting DOM XML events please emit the root element obtained from <code class="literal">document.getDocumentElement()</code>.
				</p>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-punctuation"></a>21.4.6. Data Flow Punctuation With Markers</h3>
						</div>
					</div>
				</div>
				<p>
					When your application executes a data flow instance by means of the <code class="literal">start</code> (non-blocking) or <code class="literal">run</code> (blocking) methods, the data flow instance stays running until either completed or cancelled. While cancellation is always via the <code class="literal">cancel</code> method,
					completion occurs when all source operators provide final markers.
				</p>
				<p>
					The final marker is an object that implements the <code class="literal">EPDataFlowSignalFinalMarker</code> interface. Some operators may also provide or process data window markers which implement the <code class="literal">EPDataFlowSignalWindowMarker</code> interface. All such signals implement the <code class="literal">EPDataFlowSignal</code> interface.
				</p>
				<p>
					Some source operators such as <code class="literal">EventBusSource</code> and <code class="literal">EPStatementSource</code> do not generate final markers as they act continuously.
				</p>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-api-exception"></a>21.4.7. Exception Handling</h3>
						</div>
					</div>
				</div>
				<p>
					All exceptions during the execution of a data flow are logged and reported to the <code class="literal">EPDataFlowExceptionHandler</code> instance if one was provided.
				</p>
				<p>
					If no exception handler is provided or the provided exception handler re-throws or generates a new runtime exception,
					the source operator handles the exception and completes (ends). When all source operators complete then the data flow instance transitions to complete.
				</p>
			</div>
		</div>
		<div class="sect1" lang="en-US">
			<div class="titlepage">
				<div>
					<div>
						<h2 class="title"><a id="dataflow-examples"></a>21.5. Examples</h2>
					</div>
				</div>
			</div>
			<p>
				The following example is a rolling top words count implemented as a data flow, over a 30 second time window and providing the top 3 words every 2 seconds:
			</p>
			<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow RollingTopWords
  create objectarray schema WordEvent (word string),
  
  Emitter -&gt; wordstream&lt;WordEvent&gt; {name:'a'} {} // Produces word stream
  
  Select(wordstream) -&gt; wordcount { // Sliding time window count per word
    select: (select word, count(*) as wordcount 
          from wordstream#time(30) group by word)
  }

  Select(wordcount) -&gt; wordranks { // Rank of words
    select: (select window(*) as rankedWords 
          from wordcount#sort(3, wordcount desc) 
          output snapshot every 2 seconds)
  }
  
  LogSink(wordranks) {}</pre>
			<p>
				The next example implements a bargain index computation that separates a mixed trade and quote event stream into a trade and a quote stream, computes a vwap and joins the two streams to compute an index:
			</p>
			<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow VWAPSample
  create objectarray schema TradeQuoteType as (type string, ticker string, price double, volume long, askprice double, asksize long),
  
  MyObjectArrayGraphSource -&gt; TradeQuoteStream&lt;TradeQuoteType&gt; {}
  
  Filter(TradeQuoteStream) -&gt; TradeStream {
    filter: type = "trade"
  }
  
  Filter(TradeQuoteStream) -&gt; QuoteStream {
    filter: type = "quote"
  }
  
  Select(TradeStream) -&gt; VwapTrades {
    select: (select ticker, sum(price * volume) / sum(volume) as vwap, 
          min(price) as minprice
          from TradeStream#groupwin(ticker)#length(4) group by ticker)
  }
  
  Select(VwapTrades as T, QuoteStream as Q) -&gt; BargainIndex {
    select: 
      (select case when vwap &gt; askprice then asksize * (Math.exp(vwap - askprice)) else 0.0d end as index
      from T#unique(ticker) as t, Q#lastevent as q
      where t.ticker = q.ticker)
  }
  
  LogSink(BargainIndex) {}
</pre>
			<p>
				The final example is a word count data flow, in which three custom operators tokenize, word count and aggregate. The custom operators in this example are discussed next.
			</p>
			<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow WordCount
  MyLineFeedSource -&gt; LineOfTextStream {}
  MyTokenizerCounter(LineOfTextStream) -&gt; SingleLineCountStream {}
  MyWordCountAggregator(SingleLineCountStream) -&gt; WordCountStream {}
  LogSink(WordCountStream) {}</pre>
		</div>
		<div class="sect1" lang="en-US">
			<div class="titlepage">
				<div>
					<div>
						<h2 class="title"><a id="dataflow-op"></a>21.6. Operator Implementation</h2>
					</div>
				</div>
			</div>
			<div xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="note">
				<h2>Note</h2>
				<p>
					Implementing an operator requires the use of extension and internal APIs that are not considered stable and may change between versions.
				</p>
			</div>
			<p>
				This section discusses how to implement classes that serve as operators in a data flow. The section employs the example data flow as shown earlier.
			</p>
			<p>
				This example data flow has operators <code class="literal">MyLineFeedSource</code>, <code class="literal">MyTokenizerCounter</code> and <code class="literal">MyWordCountAggregator</code> that are application provided operators:
			</p>
			<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">create dataflow WordCount
  MyLineFeedSource -&gt; LineOfTextStream {}
  MyTokenizerCounter(LineOfTextStream) -&gt; SingleLineCountStream {}
  MyWordCountAggregator(SingleLineCountStream) -&gt; WordCountStream {}
  LogSink(WordCountStream) {}</pre>
			<p>
				Each operator requires implementing the following interfaces:
			</p>
			<div class="itemizedlist">
				<ul>
					<li>
						<p>Implement the <code class="literal">DataFlowOperatorForge</code> interface for the compiler to use.</p>
					</li>
					<li>
						<p>Implement the <code class="literal">DataFlowOperatorFactory</code> interface for the runtime to instantiate operator instances.</p>
					</li>
					<li>
						<p>Implement either the <code class="literal">DataFlowOperator</code> interface, the <code class="literal">DataFlowOperatorLifecycle</code> or the <code class="literal">DataFlowSourceOperator</code> interface.</p>
					</li>
				</ul>
			</div>
			<p>
				The compiler must be able to find the class implementing <code class="literal">DataFlowOperatorForge</code>. Add the forge package or forge class to imports:
			</p>
			<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">// Sample code adds 'package.*' to simply import the package.
Configuration configuration = new Configuration();
configuration.getCommon().addImport(MyLineFeedSourceForge.class.getName());</pre>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-op-source"></a>21.6.1. Sample Operator Acting as Source</h3>
						</div>
					</div>
				</div>
				<p>
					Every operator has a forge class that implements the <code class="literal">DataFlowOperatorForge</code> interface and is only used at compile-time.
					The compiler provides the operator parameter expressions to the forge instance and invokes the <code class="literal">initializeForge</code> method.
					When it is time to compile the compiler generates code by invoking the <code class="literal">make</code> method.
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">// The OutputTypes annotation can be used to specify the type of events
// that are output by the operator.
// If provided, it is not necessary to declare output types in the data flow.
// The event representation is object-array.
@OutputTypes(value = {
        @OutputType(name = "line", typeName = "String")
})

// Provide the DataFlowOpProvideSignal annotation to indicate that
// the source operator provides a final marker.
@DataFlowOpProvideSignal
public class MyLineFeedSourceForge implements DataFlowOperatorForge {

    public DataFlowOpForgeInitializeResult initializeForge(DataFlowOpForgeInitializeContext context) throws ExprValidationException {
        return null;
    }

    public CodegenExpression make(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
        return newInstance(MyLineFeedSourceFactory.class);
    }
}</pre>
				<p>
					The operator factory class must implement the <code class="literal">DataFlowOperatorFactory</code> interface.
					At deployment time the operator factory initializes using the code generated in the forge <code class="literal">make</code> method.
					Upon instantiating a data flow the factory must return an operator instance.
				</p>
				<p>
					The implementation for the sample <code class="literal">MyLineFeedSourceFactory</code> is:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">public class MyLineFeedSourceFactory implements DataFlowOperatorFactory {

    public void initializeFactory(DataFlowOpFactoryInitializeContext context) {
    }

    public DataFlowOperator operator(DataFlowOpInitializeContext context) {
        return new MyLineFeedSource(Collections.emptyIterator());
    }
}</pre>
				<p>
					The operator implementation for the sample <code class="literal">MyLineFeedSource</code> is:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">public class MyLineFeedSource implements DataFlowSourceOperator {

    @DataFlowContext
    private EPDataFlowEmitter dataFlowEmitter;

    private final Iterator&lt;String&gt; lines;

    public MyLineFeedSource(Iterator&lt;String&gt; lines) {
        this.lines = lines;
    }

    public void open(DataFlowOpOpenContext openContext) {
    }

    public void next() {
        if (lines.hasNext()) {
            dataFlowEmitter.submit(new Object[]{lines.next()});
        } else {
            dataFlowEmitter.submitSignal(new EPDataFlowSignalFinalMarker() {
            });
        }
    }

    public void close(DataFlowOpCloseContext openContext) {
    }
}</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-op-second"></a>21.6.2. Sample Tokenizer Operator</h3>
						</div>
					</div>
				</div>
				<p>
					The implementation for the sample <code class="literal">MyTokenizerCounter</code> is a forge, factory and operator in one class:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">@OutputTypes({
        @OutputType(name = "line", type = int.class),
        @OutputType(name = "wordCount", type = int.class),
        @OutputType(name = "charCount", type = int.class)
})
public class MyTokenizerCounter implements DataFlowOperatorForge, DataFlowOperatorFactory, DataFlowOperator {
    private static final Logger log = LoggerFactory.getLogger(MyTokenizerCounter.class);

    @DataFlowContext
    private EPDataFlowEmitter graphContext;

    public DataFlowOpForgeInitializeResult initializeForge(DataFlowOpForgeInitializeContext context) throws ExprValidationException {
        return null;
    }

    public CodegenExpression make(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
        return newInstance(MyTokenizerCounter.class);
    }

    public void initializeFactory(DataFlowOpFactoryInitializeContext context) {
    }

    public DataFlowOperator operator(DataFlowOpInitializeContext context) {
        return new MyTokenizerCounter();
    }

    public void onInput(String line) {
        StringTokenizer tokenizer = new StringTokenizer(line, " \t");
        int wordCount = tokenizer.countTokens();
        int charCount = 0;
        while (tokenizer.hasMoreTokens()) {
            String token = tokenizer.nextToken();
            charCount += token.length();
        }
        log.debug("Submitting stat words[" + wordCount + "] chars[" + charCount + "] for line '" + line + "'");
        graphContext.submit(new Object[]{1, wordCount, charCount});
    }
}</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-op-third"></a>21.6.3. Sample Aggregator Operator</h3>
						</div>
					</div>
				</div>
				<p>
					The implementation for the sample <code class="literal">MyWordCountAggregator</code> with comments is:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">@OutputTypes(value = {
        @OutputType(name = "stats", type = MyWordCountStats.class)
})
public class MyWordCountAggregator implements DataFlowOperatorForge, DataFlowOperatorFactory, DataFlowOperator {
    private static final Logger log = LoggerFactory.getLogger(MyWordCountAggregator.class);

    @DataFlowContext
    private EPDataFlowEmitter graphContext;

    private final MyWordCountStats aggregate = new MyWordCountStats();

    public DataFlowOpForgeInitializeResult initializeForge(DataFlowOpForgeInitializeContext context) throws ExprValidationException {
        return null;
    }

    public CodegenExpression make(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
        return newInstance(MyWordCountAggregator.class);
    }

    public void initializeFactory(DataFlowOpFactoryInitializeContext context) {
    }

    public DataFlowOperator operator(DataFlowOpInitializeContext context) {
        return new MyWordCountAggregator();
    }

    public void onInput(int lines, int words, int chars) {
        aggregate.add(lines, words, chars);
        log.debug("Aggregated: " + aggregate);
    }

    public void onSignal(EPDataFlowSignal signal) {
        log.debug("Received punctuation, submitting totals: " + aggregate);
        graphContext.submit(aggregate);
    }
}</pre>
			</div>
			<div class="sect2" lang="en-US">
				<div class="titlepage">
					<div>
						<div>
							<h3 class="title"><a id="dataflow-op-parameters"></a>21.6.4. Passing Operator Parameters</h3>
						</div>
					</div>
				</div>
				<p>
					The forge instance receives parameters expressions. A forge can declare parameters like so:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">// Expose a parameter named "file" that takes any expression as parameter
@DataFlowOpParameter
private ExprNode file;

// Expose a parameter named "adapterInputSource" that will be an instance of some interface
// Interface implementations as parameters are declare a Map&lt;String, Object&gt;
@DataFlowOpParameter
private Map&lt;String, Object&gt; adapterInputSource;

// Expose a paramerer named "propertyNames" that is an array of string constants
@DataFlowOpParameter
private String[] propertyNames;</pre>
				<p>
					The forge class can obtain the output event type if needed. It should also validate the expression parameters and throw <code class="literal">ExprValidationException</code> if the parameter expression does not return the expected type.
					The utility class <code class="literal">DataFlowParameterValidation</code> has validate utility methods that return a validated expression:
					For example:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">public DataFlowOpForgeInitializeResult initializeForge(DataFlowOpForgeInitializeContext context) throws ExprValidationException {
  // Obtain the declared output event type
  outputEventType = context.getOutputPorts().get(0).getOptionalDeclaredType() != null ? context.getOutputPorts().get(0).getOptionalDeclaredType().getEventType() : null;
  if (outputEventType == null) {
    throw new ExprValidationException("No event type provided for output, please provide an event type name");
  }

  // validate the "file" parameter expression expected to return a String-typed value
  file = DataFlowParameterValidation.validate("file", file, String.class, context);
  return null;
}</pre>
				<p>
					The forge class passes parameters to the factory. We use <code class="literal">SAIFFInitializeBuilder</code> that is a builder utility for building the factory. For example:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">public CodegenExpression make(CodegenMethodScope parent, SAIFFInitializeSymbol symbols, CodegenClassScope classScope) {
  return new SAIFFInitializeBuilder(FileSourceFactory.class, this.getClass(), "factory", parent, symbols, classScope)
    .exprnode("file", file)
    .constant("propertyNames", propertyNames)
    .map("adapterInputSource", adapterInputSource)
    .build();
}</pre>
				<p>
					The factory class must have setter-methods of the same name that receive the parameters:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">private ExprEvaluator file;
private String[] propertyNames;
private Map&lt;String, Object&gt; adapterInputSource;

    public void setFile(ExprEvaluator file) {
        this.file = file;
    }
    
    public void setPropertyNames(String[] propertyNames) {
        this.propertyNames = propertyNames;
    }

    public void setAdapterInputSource(Map&lt;String, Object&gt; adapterInputSource) {
        this.adapterInputSource = adapterInputSource;
    }</pre>
				<p>
					The factory class can resolve parameter values
					by evaluating expressions and by determining whether parameters were passed as options. The <code class="literal">DataFlowParameterResolution</code> class provides
					convenience methods. For example:
				</p>
				<pre xmlns="" xmlns:d="http://docbook.org/ns/docbook" xmlns:rf="java:org.jboss.highlight.XhtmlRendererFactory" class="">public DataFlowOperator operator(DataFlowOpInitializeContext context) {
  String fileName = DataFlowParameterResolution.resolveWithDefault("file", file, null, String.class, context);
  AdapterInputSource adapterInputSourceInstance = DataFlowParameterResolution.resolveOptionalInstance("adapterInputSource", adapterInputSource, AdapterInputSource.class, context);
  return new MyOperator(fileName, adapterInputSourceInstance);
}</pre>
			</div>
		</div>
	</div>
	<ul xmlns:d="http://docbook.org/ns/docbook" class="docnav">
		<li class="previous"><a accesskey="p" href="spatial.html"><strong>Prev</strong>Chapter 20. EPL Reference: Spatial Methods and In...</a></li>
		<li class="up"><a accesskey="u" href="#"><strong>Top of page</strong></a></li>
		<li class="home"><a accesskey="h" href="index.html"><strong>Front page</strong></a></li>
		<li class="next"><a accesskey="n" href="extension.html"><strong>Next</strong>Chapter 22. Integration and Extension</a></li>
	</ul>
</body>

</html>